package org.example.chapter1;

import rx.Observable;
import rx.Subscriber;

import java.util.Random;

public class Demo1_3_1 {
    public static void main(String[] args) {
        createObserver().subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                System.err.println("OnCompleted");
            }

            @Override
            public void onError(Throwable throwable) {
                System.err.println("OnError");
            }

            @Override
            public void onNext(Integer integer) {
                System.err.println("Integer=" + integer);
            }
        });
    }

    private static Observable<Integer> createObserver(){
        Observable<Integer> observable = Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                if (!subscriber.isUnsubscribed()) {
                    for (int i = 0; i < 5; i++) {
                        int temp = new Random().nextInt(10);
                        if (temp > 8) {
                            subscriber.onError(new Throwable("value > 8"));
                            break;
                        }else{
                            subscriber.onNext(temp);
                        }
                        if (i == 4) {
                            subscriber.onCompleted();
                        }
                    }
                }
            }
        });
        return observable;
    }
}
